package Window_SourceCodeAnalysis;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Arrays;
/**
 * @author Spring_Hu
 * @date 2021/10/13 22:25
 */
public class Window_SourceCodeAnalysis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
       // env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] line = value.split(",");
                        return new WaterSensor(
                                line[0],
                                Long.parseLong(line[1]),
                                Integer.parseInt(line[2]));
                    }
                });


        WindowedStream<WaterSensor, String, TimeWindow> sensorWS = sensorDS
                .keyBy(sensor -> sensor.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));   // 窗口分配器

        // TODO 全窗口函数
        SingleOutputStreamOperator<String> resultDS = sensorWS.process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                out.collect("key=" + s + "\n"
                        + "窗口有" + elements.spliterator().estimateSize() + "条数据" + "\n"
                        + "数据为：" + Arrays.asList(elements).toString() + "\n"
                        + "窗口划分:[" + context.window().getStart() + "," + context.window().getEnd() + ")" + "\n\n"
                );
            }
        });

        resultDS.print();


        env.execute();
    }
}
/*
    以滚动窗口为例：(全局窗口是自定义的globaltime（不是基于时间的窗口） 是不知道何时开始何时结束 不指定触发器就不会触发)
    1、窗口是怎么划分的？

    offset为调整时差的 参数可以用0代替

       long start =
                TimeWindow.getWindowStartWithOffset(
                        now, (globalOffset + staggerOffset) % size, size);
        return Collections.singletonList(new TimeWindow(start, start + size));

            public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
        return timestamp - (timestamp - offset + windowSize) % windowSize;
    }

            start = timestamp - (timestamp - offset + windowSize) % windowSize;
                =>  start实现的效果，当前时间（要么是事件事件要么是处理时间 ）向下取窗口长度的整数倍（时间戳）
            end = start + size

    2、窗口为什么是左闭右开？

        Gets the largest timestamp that still belongs to this window.

    <p>This timestamp is identical to {@code getEnd() - 1}.

    @return The largest timestamp that still belongs to this window.

        public long maxTimestamp() {
        return end - 1;
    }
            maxTimestamp = end - 1ms

    3、窗口什么时候触发？

    根据源码中可以发现 找到触发器的 getDefaultTrigger方法
         ctx.registerProcessingTimeTimer(window.maxTimestamp());

             时间大于等于最大时间戳这个窗口就要被触发 即输出结果
            时间 >= maxTimestamp（end - 1ms）

    4、窗口什么时候创建？窗口什么时候销毁？（窗口的生命周期？）

            return Collections.singletonList(new TimeWindow(start, start + size));

            创建： 属于本窗口的第一条数据来的时候，new的，放到一个 单例集合中
            销毁： 7.6之前讲

        行业中称创建的窗口为 桶

     /**·
 * TODO Watermark理解
 *   1、用来衡量事件时间的进展
 *   2、用来处理乱序问题
 *   3、是一个特殊的时间戳，从生成位置插入到流里面
 *   4、单调不减的
 *   5、用来触发窗口等
 *   6、Flink认为，事件时间小于watermark的数据应该都已经处理过了
 *      如果后续还有事件时间小于watermark的数据来，称为迟到数据
 *
 *  TODO Watermark的生成方式分类
 *    1、周期性（onPeriodicEmit里发射watermark）: 默认200ms
 *    2、间歇性也叫打点 即每条数据都生成（onEvent里发射watermark）
 *
 *  TODO Watermark的写法、生成逻辑
 *    1、升序写法：forMonotonousTimestamps
 *          watermark = maxTs - 1ms
 *    2、乱序写法：forBoundedOutOfOrderness(Duration)
 *          watermark = maxTs - Duration - 1ms
 *
 *  TODO Watermark的传递
 *    从指定位置插入的，向下游传递
 *      一对多：广播                 发送者
 *      多对一：取最小（木桶原理）    接收者
 *      多对多：上面两者的结合，拆开来看
 *
 *  TODO Flink对乱序和迟到数据的处理方式
 *     1、Watermark处理乱序
 *     2、窗口允许迟到
 *     3、侧输出流
 *
 */


